
package client

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"math/rand"
	"net"
	"net/http"
	"net/url"
	"sort"
	"strconv"
	"sync"
	"time"

	"go.etcd.io/etcd/api/v3/version"
)

var (
	ErrNoEndpoints           = errors.New("client: no endpoints available")
	ErrTooManyRedirects      = errors.New("client: too many redirects")
	ErrClusterUnavailable    = errors.New("client: etcd cluster is unavailable or misconfigured")
	ErrNoLeaderEndpoint      = errors.New("client: no leader endpoint available")
	errTooManyRedirectChecks = errors.New("client: too many redirect checks")

	// oneShotCtxValue是在使用WithValue（&oneShotValue）的上下文上设置的，因此
	// Do（）不会重试请求
	oneShotCtxValue interface{}
)

var DefaultRequestTimeout = 5 * time.Second

var DefaultTransport CancelableTransport = &http.Transport{
	Proxy: http.ProxyFromEnvironment,
	DialContext: (&net.Dialer{
		Timeout:   30 * time.Second,
		KeepAlive: 30 * time.Second,
	}).DialContext,
	TLSHandshakeTimeout: 10 * time.Second,
}

type EndpointSelectionMode int

const (
	// EndpointSelectionRandom是“SelectionMode”的默认值。顾名思义，客户机对象将以随机方式从集群的成员
	// 中选择一个节点。如果集群有三个成员：A、B、
	// 和C，则客户端从其三个成员中选择任意节点作为其请求
	// 目的地。
	EndpointSelectionRandom EndpointSelectionMode = iota

	// 如果将“SelectionMode”设置为“EndpointSelectionPriorityLeader”，则
	// 请求将直接发送到集群领导。与向etcd跟随者发出请求相比，这减少了
	// 转发往返，etcd跟随者随后将请求转发给集群领导者。但是，如果某个引导者
	// 失败，则以这种方式配置的客户端无法在
	// 其余etcd跟随者之间进行优先级排序。因此，当客户端将'SelectionMode'
	// 设置为'EndpointSelectionPriorityLeader'时，它必须使用'client'。AutoSync（）'到
	// 保持其对当前群集状态的了解。
	// 
	// 此模式应与客户端一起使用。自动同步（）。
	EndpointSelectionPrioritizeLeader
)

type Config struct {
	// 端点定义了一组URL（仅限方案、主机和端口）
	// 可用于与逻辑etcd群集通信。对于
	// 例如，可以提供如下三节点集群：
	// 
	// 端点：[]字符串{
	// “http:
	// ”http:
	// ”http:
	// 
	// 
	// 如果提供了多个终结点，客户端将尝试在一个或多个终结点不可用的情况下使用所有终结点。
	// 
	// 如果调用了Client.Sync，客户端可能会缓存一组备用的终结点以继续operation.
	Endpoints []string

	// 传输被客户端用来驱动HTTP请求。如果没有提供
	// 将使用DefaultTransport。
	Transport CancelableTransport

	// CheckRedirect指定处理HTTP重定向的策略。
	// 如果CheckRedirect不为零，客户端将在
	// follo之前调用它使用HTTP重定向。唯一的参数是已经发出的
	// 请求数。如果CheckRedirect返回
	// 错误，客户端。Do将不会发出任何进一步的请求，并返回
	// 错误返回给调用方。
	// 
	// 如果选中了CheckRedirect无，客户端使用其默认策略
	// 在连续10次请求后停止。
	CheckRedirect CheckRedirectFunc

	// Username指定要添加为授权头的用户凭据
	Username string

	// Password是指定用户要添加为请求的授权头
	// 的密码。
	Password string

	// HeaderTimeoutPerRequest指定在客户端发出的单个请求中等待响应的时间限制
	// 标头。超时包括
	// 连接时间、任何重定向和标头等待时间。
	// 
	// 对于非监视GET请求，服务器返回r立即响应正文。对于PUT/POST/DELETE请求，服务器将在响应之前尝试提交请求，这将花费'100ms+2*RTT`。对于监视请求，服务器将立即返回头以通知客户端
	// 监视启动。但如果服务器是在某种代理背后，响应
	// 头可能缓存在代理上，客户端无法依赖此行为。
	// 
	// 特别是，等待请求将忽略此超时。
	// 
	// 一个API调用可能会向不同的etcd服务器发送多个请求，直到超时为止卡介苗
	// 成功了。使用API的上下文指定总超时。
	// 
	// HeaderTimeoutPerRequest为零表示没有超时。
	HeaderTimeoutPerRequest time.Duration

	// SelectionMode是一个EndpointSelectionMode枚举，指定用于选择向其发送请求的etcd群集节点的
	// 策略。
	SelectionMode EndpointSelectionMode
}

func (cfg *Config) transport() CancelableTransport {
	if cfg.Transport == nil {
		return DefaultTransport
	}
	return cfg.Transport
}

func (cfg *Config) checkRedirect() CheckRedirectFunc {
	if cfg.CheckRedirect == nil {
		return DefaultCheckRedirect
	}
	return cfg.CheckRedirect
}

// 可取消传输模拟net/http。传输，但要求
// 对象也支持请求取消。
type CancelableTransport interface {
	http.RoundTripper
	CancelRequest(req *http.Request)
}

type CheckRedirectFunc func(via int) error

// DefaultCheckRedirect最多可执行10次重定向，但不能再执行更多。
var DefaultCheckRedirect CheckRedirectFunc = func(via int) error {
	if via > 10 {
		return ErrTooManyRedirects
	}
	return nil
}

type Client interface {
	// 同步更新etcd群集成员的内部缓存。
	Sync(context.Context) error

	// AutoSync每隔给定的时间间隔定期调用Sync（）。
	// 建议的同步间隔为10秒到1分钟，这不会给服务器带来太多开销，并使客户端及时赶上
	// 集群更改。
	// 
	// 使用它的示例：
	// 
	// 对于{
	// /err:=client.AutoSync（ctx，10*time.Second）
	// 如果err==context.deadlineextended | err==context.cancelled{
	// log.Print（err）
	// }
	AutoSync(context.Context, time.Duration) error

	// 端点返回客户端用于解析HTTP请求的当前API端点集的副本。如果调用过Sync，则
	// 这可能与配置中提供的初始端点不同。
	Endpoints() []string

	// SetEndpoints设置客户端用于解析
	// HTTP请求的API端点集。如果给定的终结点无效，则会出现错误
	// 返回
	SetEndpoints(eps []string) error

	// GetVersion检索当前etcd服务器和群集版本
	GetVersion(ctx context.Context) (*version.Versions, error)

	httpClient
}

func New(cfg Config) (Client, error) {
	c := &httpClusterClient{
		clientFactory: newHTTPClientFactory(cfg.transport(), cfg.checkRedirect(), cfg.HeaderTimeoutPerRequest),
		rand:          rand.New(rand.NewSource(int64(time.Now().Nanosecond()))),
		selectionMode: cfg.SelectionMode,
	}
	if cfg.Username != "" {
		c.credentials = &credentials{
			username: cfg.Username,
			password: cfg.Password,
		}
	}
	if err := c.SetEndpoints(cfg.Endpoints); err != nil {
		return nil, err
	}
	return c, nil
}

type httpClient interface {
	Do(context.Context, httpAction) (*http.Response, []byte, error)
}

// httpClusterClient.clientFactory 字段指向了一个 newHTTPClientFactory（）工厂函数，该函数负
// 责根据指定的 ClientURL 创建 httprClient 实例
func newHTTPClientFactory(tr CancelableTransport, cr CheckRedirectFunc, headerTimeout time.Duration) httpClientFactory {
	return func(ep url.URL) httpClient {	// 工厂 函数
		return &redirectFollowingHTTPClient{	// 创建 redirectFollowingHTTPClient 实例
			checkRedirect: cr,
			client: &simpleHTTPClient{
				transport:     tr,
				endpoint:      ep,
				headerTimeout: headerTimeout,
			},
		}
	}
}

type credentials struct {
	username string
	password string
}

type httpClientFactory func(url.URL) httpClient

type httpAction interface {
	HTTPRequest(url.URL) *http.Request
}

type httpClusterClient struct {
	clientFactory httpClientFactory	// 工厂函数，用于创建底层 的 httpClient 实例 。
	endpoints     []url.URL	// 记录集群中所有节点 的暴露给客户端的 URL 地址。
	pinned        int	// 用于选择重试的 URL 地址。
	credentials   *credentials
	sync.RWMutex
	rand          *rand.Rand	// 随机数，用于选择重试的 URL 地址
	// 更新 endpoints 宇段的模式，目前有两
	// 个可选值，分别是 EndpointSelectionRandom 和 EndpointSelectionPrioritizeLeader 。
	selectionMode EndpointSelectionMode
}

func (c *httpClusterClient) getLeaderEndpoint(ctx context.Context, eps []url.URL) (string, error) {
	ceps := make([]url.URL, len(eps))
	copy(ceps, eps)

	// 以在不使用当前
	// 客户端的情况下对新终结点列表执行查找，我们将复制它
	clientCopy := &httpClusterClient{
		clientFactory: c.clientFactory,
		credentials:   c.credentials,
		rand:          c.rand,

		pinned:    0,
		endpoints: ceps,
	}

	mAPI := NewMembersAPI(clientCopy)
	leader, err := mAPI.Leader(ctx)
	if err != nil {
		return "", err
	}
	if len(leader.ClientURLs) == 0 {
		return "", ErrNoLeaderEndpoint
	}

	return leader.ClientURLs[0], nil // TODO:如何处理多个客户端URL？
}

func (c *httpClusterClient) parseEndpoints(eps []string) ([]url.URL, error) {
	if len(eps) == 0 {
		return []url.URL{}, ErrNoEndpoints
	}

	neps := make([]url.URL, len(eps))
	for i, ep := range eps {
		u, err := url.Parse(ep)
		if err != nil {
			return []url.URL{}, err
		}
		neps[i] = *u
	}
	return neps, nil
}

func (c *httpClusterClient) SetEndpoints(eps []string) error {
	neps, err := c.parseEndpoints(eps)
	if err != nil {
		return err
	}

	c.Lock()
	defer c.Unlock()

	c.endpoints = shuffleEndpoints(c.rand, neps)
	// 我们在这里没有为你做任何事。这是
	// 由于没有上下文，我们无法调用getLeaderEndpoint 
	// 但是，如果您使用的是PrioritizeLeader，您已经被告知
	// 定期调用sync，在这里我们有一个ctx，并且可以计算出
	// leader。PrioritizeLeader也是一个非常松散的保证，因此处理它当端点列表中的项发生变化时，确保这是一个无leader响应，我们选择了一个新的管脚
	c.pinned = 0

	return nil
}

// 从 endpoints 字段中选择合适的 URL 地址并
// 建立连接，然后发送相应的请求
func (c *httpClusterClient) Do(ctx context.Context, act httpAction) (*http.Response, []byte, error) {
	action := act
	c.RLock()
	leps := len(c.endpoints)
	// 复制目前集群提供的 URL 地址
	eps := make([]url.URL, leps)
	n := copy(eps, c.endpoints)
	pinned := c.pinned

	if c.credentials != nil {
		action = &authedAction{
			act:         act,
			credentials: *c.credentials,
		}
	}
	c.RUnlock()

	if leps == 0 {
		return nil, nil, ErrNoEndpoints
	}

	if leps != n {
		return nil, nil, errors.New("unable to pick endpoint: copy failed")
	}

	var resp *http.Response
	var body []byte
	var err error
	cerr := &ClusterError{}
	isOneShot := ctx.Value(&oneShotCtxValue) != nil

	for i := pinned; i < leps+pinned; i++ {
		k := i % leps
		hc := c.clientFactory(eps[k])	// 从 eps 中选择一个 URL 地址创建连接
		resp, body, err = hc.Do(ctx, action)	// 调用 httpClient.Do （）方法向服务端发送请求
		if err != nil {
			cerr.Errors = append(cerr.Errors, err)
			if err == ctx.Err() {
				return nil, nil, ctx.Err()
			}
			if err == context.Canceled || err == context.DeadlineExceeded {
				return nil, nil, err
			}
		} else if resp.StatusCode/100 == 5 {
			// 处理 5xx 的错误码
			switch resp.StatusCode {
			case http.StatusInternalServerError, http.StatusServiceUnavailable:
				cerr.Errors = append(cerr.Errors, fmt.Errorf("client: etcd member %s has no leader", eps[k].String()))
			default:
				cerr.Errors = append(cerr.Errors, fmt.Errorf("client: etcd member %s returns server error [%s]", eps[k].String(), http.StatusText(resp.StatusCode)))
			}
			err = cerr.Errors[0]
		}
		if err != nil {
			if !isOneShot {
				continue
			}
			c.Lock()
			c.pinned = (k + 1) % leps
			c.Unlock()
			return nil, nil, err
		}
		if k != pinned {
			c.Lock()
			c.pinned = k
			c.Unlock()
		}
		return resp, body, nil
	}

	return nil, nil, cerr
}

func (c *httpClusterClient) Endpoints() []string {
	c.RLock()
	defer c.RUnlock()

	eps := make([]string, len(c.endpoints))
	for i, ep := range c.endpoints {
		eps[i] = ep.String()
	}

	return eps
}

// Client v2 中也有与 Client v3 类似的同步机制，其主要实现在 httpC!usterC!ient.Sync （）方法
// 中 。 Sync（） 方法首先会请求当前集群中所有节点提供的 ClientURL 地址， 然后根据当前
// selectionMode 字段指定的模式修改 pinned 字段值， 最后更新 endpoints 字段中记录的 ClientURL
// 地址
func (c *httpClusterClient) Sync(ctx context.Context) error {
	mAPI := NewMembersAPI(c)
	ms, err := mAPI.List(ctx)	// 请求集群中各个节点的信息
	if err != nil {
		return err
	}

	var eps []string
	for _, m := range ms {	// 将获取到的 URL 地址添加到 eps 数量且中
		eps = append(eps, m.ClientURLs...)
	}

	neps, err := c.parseEndpoints(eps)	// 解析 URL ，
	if err != nil {
		return err
	}

	npin := 0

	switch c.selectionMode {
	case EndpointSelectionRandom:
		c.RLock()
		eq := endpointsEqual(c.endpoints, neps)	// 比较 Client URL 地址是否变化
		c.RUnlock()

		if eq {
			return nil
		}
		neps = shuffleEndpoints(c.rand, neps)	// 洗牌，打舌L URL 顺序
	case EndpointSelectionPrioritizeLeader:
		nle, err := c.getLeaderEndpoint(ctx, neps)	// 获取 Leader 提供的 ClientURL ，
		if err != nil {
			return ErrNoLeaderEndpoint
		}

		for i, n := range neps {	// 边历全部的 URL
			if n.String() == nle {
				npin = i	// 将 npin 指向 Leader 提供的 ClientURL 地址的下标
				break
			}
		}
	default:
		return fmt.Errorf("invalid endpoint selection mode: %d", c.selectionMode)
	}

	c.Lock()
	defer c.Unlock()
	c.endpoints = neps	// 	更新 endpoints 字段
	c.pinned = npin	// 更新 pinned 字段

	return nil
}

func (c *httpClusterClient) AutoSync(ctx context.Context, interval time.Duration) error {
	ticker := time.NewTicker(interval)
	defer ticker.Stop()
	for {
		err := c.Sync(ctx)
		if err != nil {
			return err
		}
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-ticker.C:
		}
	}
}

func (c *httpClusterClient) GetVersion(ctx context.Context) (*version.Versions, error) {
	act := &getAction{Prefix: "/version"}

	resp, body, err := c.Do(ctx, act)
	if err != nil {
		return nil, err
	}

	switch resp.StatusCode {
	case http.StatusOK:
		if len(body) == 0 {
			return nil, ErrEmptyBody
		}
		var vresp version.Versions
		if err := json.Unmarshal(body, &vresp); err != nil {
			return nil, ErrInvalidJSON
		}
		return &vresp, nil
	default:
		var etcdErr Error
		if err := json.Unmarshal(body, &etcdErr); err != nil {
			return nil, ErrInvalidJSON
		}
		return nil, etcdErr
	}
}

type roundTripResponse struct {
	resp *http.Response
	err  error
}

type simpleHTTPClient struct {
	// 内嵌了 http.RoundTripper ，用于发送 HTTP请求并获取相应的响应。
	transport     CancelableTransport
	// 请求的 URL 地址
	endpoint      url.URL
	// 请求的超时时间。
	headerTimeout time.Duration
}

// errnoreRequest表示找不到HTTPRequest对象
// 或为零。无法继续处理。
var ErrNoRequest = errors.New("no HTTPRequest was available")

func (c *simpleHTTPClient) Do(ctx context.Context, act httpAction) (*http.Response, []byte, error) {
	req := act.HTTPRequest(c.endpoint)	// 创建 Request 实例
	if req == nil {
		return nil, nil, ErrNoRequest
	}

	if err := printcURL(req); err != nil {
		return nil, nil, err
	}
	// 设置起时时间
	isWait := false
	if req.URL != nil {
		ws := req.URL.Query().Get("wait")
		if len(ws) != 0 {
			var err error
			isWait, err = strconv.ParseBool(ws)
			if err != nil {
				return nil, nil, fmt.Errorf("wrong wait value %s (%v for %+v)", ws, err, req)
			}
		}
	}

	var hctx context.Context
	var hcancel context.CancelFunc
	if !isWait && c.headerTimeout > 0 {
		hctx, hcancel = context.WithTimeout(ctx, c.headerTimeout)
	} else {
		hctx, hcancel = context.WithCancel(ctx)
	}
	defer hcancel()

	reqcancel := requestCanceler(c.transport, req)

	rtchan := make(chan roundTripResponse, 1)
	go func() {	// 启动一个后台 goroutine 发送请求
		resp, err := c.transport.RoundTrip(req)
		// 将响应写入 rtchan 通道 中
		rtchan <- roundTripResponse{resp: resp, err: err}
		close(rtchan)
	}()

	var resp *http.Response
	var err error

	select {
	case rtresp := <-rtchan:	// 监听 rtchan 通道获取响应
		resp, err = rtresp.resp, rtresp.err
	case <-hctx.Done():
		// 在继续之前取消并等待请求实际退出
		reqcancel()
		rtresp := <-rtchan
		resp = rtresp.resp
		switch {
		case ctx.Err() != nil:
			err = ctx.Err()
		case hctx.Err() != nil:
			err = fmt.Errorf("client: endpoint %s exceeded header timeout", c.endpoint.String())
		default:
			panic("failed to get error from context")
		}
	}

	// 始终检查是否有响应以处理可能的
	// 
	defer func() {
		if resp != nil {
			resp.Body.Close()
		}
	}()

	if err != nil {
		return nil, nil, err
	}

	var body []byte
	done := make(chan struct{})
	go func() {	// 启动一个后台 goroutine 读取响应的数据
		body, err = io.ReadAll(resp.Body)
		// 读取完成后，向 done 通道发送信号
		done <- struct{}{}
	}()

	select {
	case <-ctx.Done():
		resp.Body.Close()
		<-done	// 阻塞等待后台 goroutine 读取响应体结束
		return nil, nil, ctx.Err()
	case <-done:
	}

	return resp, body, err
}

type authedAction struct {
	act         httpAction
	credentials credentials
}

func (a *authedAction) HTTPRequest(url url.URL) *http.Request {
	r := a.act.HTTPRequest(url)
	r.SetBasicAuth(a.credentials.username, a.credentials.password)
	return r
}

type redirectFollowingHTTPClient struct {
	client        httpClient
	checkRedirect CheckRedirectFunc
}

// Do（） 方法主要处理 Redirect 跳转的响应 ，
func (r *redirectFollowingHTTPClient) Do(ctx context.Context, act httpAction) (*http.Response, []byte, error) {
	next := act
	for i := 0; i < 100; i++ {
		// 检测当前 Redirect 的次数， 是否还能继续 Redirect
		if i > 0 {
			if err := r.checkRedirect(i); err != nil {
				return nil, nil, err
			}
		}
		resp, body, err := r.client.Do(ctx, next)
		if err != nil {
			return nil, nil, err
		}
		if resp.StatusCode/100 == 3 {	// 处理 3XX 的响反
			hdr := resp.Header.Get("Location")	// 获取响反的 Location
			if hdr == "" {
				return nil, nil, fmt.Errorf("location header not set")
			}
			loc, err := url.Parse(hdr)	// 解析 Location 指定的地址
			if err != nil {
				return nil, nil, fmt.Errorf("location header not valid URL: %s", hdr)
			}
			next = &redirectedHTTPAction{	// 重新创建 redirectedHTTPAction 实例，并重试
				action:   act,
				location: *loc,
			}
			continue
		}
		return resp, body, nil
	}

	return nil, nil, errTooManyRedirectChecks
}

type redirectedHTTPAction struct {
	action   httpAction
	location url.URL
}

func (r *redirectedHTTPAction) HTTPRequest(ep url.URL) *http.Request {
	orig := r.action.HTTPRequest(ep)
	orig.URL = &r.location
	return orig
}

func shuffleEndpoints(r *rand.Rand, eps []url.URL) []url.URL {
	// 以上通道之间的竞争条件复制自Go 1.9<=兰德。兰德。Perm-
	n := len(eps)
	p := make([]int, n)
	for i := 0; i < n; i++ {
		j := r.Intn(i + 1)
		p[i] = p[j]
		p[j] = i
	}
	neps := make([]url.URL, n)
	for i, k := range p {
		neps[i] = eps[k]
	}
	return neps
}

func endpointsEqual(left, right []url.URL) bool {
	if len(left) != len(right) {
		return false
	}

	sLeft := make([]string, len(left))
	sRight := make([]string, len(right))
	for i, l := range left {
		sLeft[i] = l.String()
	}
	for i, r := range right {
		sRight[i] = r.String()
	}

	sort.Strings(sLeft)
	sort.Strings(sRight)
	for i := range sLeft {
		if sLeft[i] != sRight[i] {
			return false
		}
	}
	return true
}
